package com.yungu.swift.socket.component;


import com.yungu.swift.socket.communication.message.PushMessage;
import com.yungu.swift.socket.communication.message.base.Message;
import com.yungu.swift.socket.push.model.PushContent;
import com.yungu.swift.utils.JsonUtils;
import com.yungu.swift.utils.StringUtils;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.*;

/**
 * ChannelKeeper
 *
 * @author cuixiuyin
 * @date 2018/12/14
 */
@Slf4j
public class ChannelKeeper {
    /**
     * 存储每个连接
     */
    private static ChannelCache<String, Channel> CACHE = new ChannelCache<>();

    /**
     * 添加连接到容器
     *
     * @param uuid
     * @param channel
     */
    public static void add(String uuid, Channel channel) {
        log.debug("channelKeeper add:" + uuid + "-->" + channel.id());
        if (StringUtils.isNull(uuid)) {
            return;
        }
        CACHE.put(uuid, channel);
    }

    /**
     * 获取连接
     *
     * @param uuid
     * @return
     */
    public static Channel get(String uuid) {
        log.info("channelKeeper get:" + uuid);
        if (StringUtils.isNull(uuid)) {
            return null;
        }
        return CACHE.get(uuid);
    }

    /**
     * 移除连接
     *
     * @param channel
     */
    public static void remove(Channel channel) {
        log.debug("channelKeeper remove:" + channel.id());
        CACHE.entrySet().stream().filter(entry -> entry.getValue().equals(channel)).forEach(entry -> {
            log.debug("channelKeeper remove find uuid-->" + entry.getKey());
            CACHE.remove(entry.getKey());
        });
    }

    public static void removeByUuid(String uuid) {
        log.debug("channelKeeper remove:" + uuid);
        if (StringUtils.isNull(uuid)) {
            return;
        }
        CACHE.remove(uuid);
    }

    /**
     * 发送
     *
     * @param message
     * @param channel
     */
    private static void sendMessage(Channel channel, Message message) {
        TextWebSocketFrame textWebSocketFrame = new TextWebSocketFrame(JsonUtils.toJSONString(message));
        channel.writeAndFlush(textWebSocketFrame);
        ReferenceCountUtil.release(message);
        log.info("send msg ok by >> {}", message);
    }

    /**
     * 推送消息
     *
     * @param message
     */
    public static void sendMessage(String uuid, Message message) {
        log.info("当前推送对象：{},{}",uuid,JsonUtils.toJSONString(message));
        if (StringUtils.isNull(uuid)) {
            return;
        }
        Channel channel = get(uuid);
        if (channel != null) {
            log.info("获得推送通道，并执行推送，推送对象为：{},{}",uuid,JsonUtils.toJSONString(message));
            TextWebSocketFrame textWebSocketFrame = new TextWebSocketFrame(JsonUtils.toJSONString(message));
            channel.writeAndFlush(textWebSocketFrame);
            ReferenceCountUtil.release(message);
        }
    }

    /**
     * 推送消息
     *
     * @param uuid
     * @param message
     */
    public static void sendMessage(String uuid, String message) {
        PushMessage pushMessage = new PushMessage(uuid, message);
        sendMessage(uuid, pushMessage);
    }

    /**
     * 批量推送消息（按前缀）
     *
     * @param prefix
     * @param separator
     * @param message
     */
    public static void batchSendMessageByPrefix(String prefix, String separator, PushContent message) {
        Collection<Map.Entry<String, Channel>> channels = getAllByPrefix(prefix);
        if (null != channels && !channels.isEmpty()) {
            //去重map
            Map sysUserMap = new HashMap();
            for (Map.Entry<String, Channel> channelEntry : channels) {
                String uuid = channelEntry.getKey().split(separator)[1];
                if (!sysUserMap.containsKey(uuid)) {
                    sendMessage(channelEntry.getValue(), new PushMessage(channelEntry.getKey(), JsonUtils.toJSONString(message)));
                    sysUserMap.put(uuid, true);
                }
            }
        } else {
            log.warn("----- 获取通道失败,prefix:{} -----", prefix);
        }
    }

    /**
     * 批量推送消息（按前缀 - 不去重、全发）
     *
     * @param prefix
     * @param message
     */
    public static void batchSendMessageAllByPrefix(String prefix, PushContent message) {
        Collection<Map.Entry<String, Channel>> channels = getAllByPrefix(prefix);
        if (null != channels && !channels.isEmpty()) {
            for (Map.Entry<String, Channel> channelEntry : channels) {
                sendMessage(channelEntry.getValue(), new PushMessage(channelEntry.getKey(), JsonUtils.toJSONString(message)));
            }
        } else {
            log.warn("----- 获取通道失败 ----- ");
        }
    }


    /**
     * 获取全部连接（带前缀）
     *
     * @return
     */
    public static Collection<Map.Entry<String, Channel>> getAllByPrefix(String prefix) {
        List<Map.Entry<String, Channel>> channels = new ArrayList<>();
        for (Map.Entry<String, Channel> entry : CACHE.entrySet()) {
            if (entry.getKey().indexOf(prefix) == 0) {
                channels.add(entry);
            }
        }
        return channels;
    }

}
